package com.jinx.zookeeper.lock;

import org.apache.zookeeper.AddWatchMode;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.Arrays;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 *  zookeeper方式实现分布式锁
 * @author Jinx
 * @date 2021/6/14 17:36
 */
public class ZookeeperDistributedLock implements DistributedLock{


    public ZooKeeper zooKeeper;

    {
        try {
            zooKeeper = new ZooKeeper("192.168.153.128", 60000, null);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private final String path = "/zookeeperLock";


    @Override
    public void lock(){
        sureEstablished();
        final String thread = String.valueOf(Thread.currentThread().getId());
        try {
            if(zooKeeper.exists(path, null) == null){
                try {
                    zooKeeper.create(path, thread.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
                    System.out.println(Thread.currentThread().getName() + " \t 获取锁成功");
                }catch (InterruptedException | KeeperException interruptedException) {
                    // 抢占失败可能是别人比他快  或者连接失效   递归重新抢
                    System.out.println(Thread.currentThread().getName() + " \t 获取锁失败    上面");
                    this.lock();
                }

            }else{
                System.out.println(Thread.currentThread().getName() + " \t 获取锁失败   下面");
                // 没抢到  看下当前持有锁的是不是自己
                if(Arrays.toString(zooKeeper.getData(path, null, new Stat())).equals(thread)){
                    return;
                }

                // 需要等待
                CountDownLatch cdl = new CountDownLatch(1);
                this.addWatcher(cdl);

                // 等待  醒了后递归
                cdl.await();
                this.lock();
            }
        } catch (InterruptedException | KeeperException ignored) {

        }
    }



    @Override
    public void unLock() {
        sureEstablished();

        try {
            zooKeeper.close();
            System.out.println(Thread.currentThread().getName() + " \t 释放锁成功");
        } catch (InterruptedException ignored) {

        }
    }


    /**
     *  递归添加watcher
     */
    private void addWatcher(CountDownLatch cdl){
        try {
            zooKeeper.addWatch(path, watchedEvent -> {
                if (watchedEvent.getType() == Watcher.Event.EventType.NodeDeleted) {
                    cdl.countDown();
                }else {
                    // 其他的事件不处理  重新添加事件
                    this.addWatcher(cdl);
                }
            }, AddWatchMode.PERSISTENT);
        } catch (KeeperException | InterruptedException ignored) {

        }
    }

    /**
     *  是否已经建立好连接
     */
    public void sureEstablished(){
        while (zooKeeper.getState() != ZooKeeper.States.CONNECTED){
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
